feat: add on-demand security-contacts ingest for a single purl (CM-1313) - #4312
Conversation
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
|
|
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
This PR adds an on-demand path for ingesting security contacts for a single package (purl). Previously, security contacts were only populated by a daily batch sweep restricted to critical, GitHub-hosted repos. Now, when the public getPackage endpoint serves a package whose contacts have never been refreshed, it synchronously drives a Temporal workflow to ingest contacts for that package's best repo, then re-reads and returns the fresh detail. Shared repo-processing logic is extracted from the batch path so both paths reuse the same extractor pipeline.
Changes:
- New
ingestSingle.tsselects the same "best" GitHub repo the read side surfaces (nois_criticalfilter) and runs the extractor pipeline for that single repo, plus a new activity and short-timeout workflow (ingestSecurityContactsForPurlWorkflow) for synchronous API use. getPackage.tstriggers the on-demand workflow (deterministic SHA-1 workflow id +USE_EXISTINGconflict policy to dedupe concurrent requests) whencontactsLastRefreshedis null, then re-reads package detail; failures are caught and cached detail is served.- Refactored
processBatch.tsto export sharedbuildBaseDeps/processRepo; wired new exports through the activities/workflows barrels.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
backend/src/api/public/v1/packages/getPackage.ts |
Adds synchronous on-demand ingest trigger + deterministic workflow-id helper on the public package-detail endpoint. |
services/apps/packages_worker/src/security-contacts/ingestSingle.ts |
New single-purl ingest: best-repo selection SQL + extractor pipeline invocation. |
services/apps/packages_worker/src/security-contacts/workflows.ts |
Adds a short-timeout activity proxy and ingestSecurityContactsForPurlWorkflow. |
services/apps/packages_worker/src/security-contacts/activities.ts |
New ingestSecurityContactsForPurlActivity wrapping the single-purl ingest. |
services/apps/packages_worker/src/security-contacts/processBatch.ts |
Extracts and exports buildBaseDeps / processRepo for reuse. |
services/apps/packages_worker/src/activities.ts |
Exports the new activity for worker registration. |
services/apps/packages_worker/src/workflows/index.ts |
Exports the new workflow for worker registration. |
Key review notes:
- The on-demand guard (
contactsLastRefreshed == null) will re-fire on every request for packages whose best repo is non-GitHub or absent, since nothing ever marks those as attempted — turning a public read into a repeated no-op Temporal round-trip. - The synchronous
executecall has noworkflowExecutionTimeout/scheduleToStartTimeout, so the public request can hang if the worker/queue is unavailable. - A doc comment references a non-existent
security-contacts.md.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| await req.temporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', { | ||
| taskQueue: 'packages-worker', | ||
| workflowId: ondemandWorkflowId(purl), | ||
| workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, | ||
| workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, | ||
| args: [purl], | ||
| }) |
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
| workflowId: ondemandWorkflowId(purl), | ||
| workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, | ||
| workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, | ||
| args: [purl], |
There was a problem hiding this comment.
Shared repo concurrent ingest race
Medium Severity
On-demand workflow IDs are hashed per purl, so two packages sharing the same GitHub repo can run separate workflows at once. Both call processRepo / writeContacts on the same repoId, while writeContacts assumes a single writer per repo.
Reviewed by Cursor Bugbot for commit 471ad1d. Configure here.
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
| if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { | ||
| throw new Error( | ||
| 'Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE', | ||
| ) | ||
| } |
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
| workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, | ||
| workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, | ||
| args: [purl], | ||
| }) |
There was a problem hiding this comment.
Concurrent repo contact writes
Medium Severity
On-demand ingest can run processRepo for a repo while the scheduled batch sweep processes that same repo in parallel. writeContacts assumes a single writer per repoId and soft-deletes then upserts without locking, so overlapping runs can interleave and produce inconsistent security_contacts or repo policy fields.
Reviewed by Cursor Bugbot for commit 59fdeee. Configure here.
| try { | ||
| await processRepo(target, baseDeps, qx) | ||
| } catch (err) { | ||
| log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed') | ||
| await markRepoAttempted(qx, target.repoId).catch(() => undefined) | ||
| } |
| await packagesTemporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', { | ||
| taskQueue: 'packages-worker', | ||
| workflowId: ondemandWorkflowId(purl), | ||
| workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, | ||
| workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING, | ||
| args: [purl], | ||
| }) |
| throw new NotFoundError() | ||
| } | ||
|
|
||
| if (pkg.contactsLastRefreshed == null) { |
There was a problem hiding this comment.
Indeed, it makes sense!
Will keep it in mind when implementing the security contacts endpoint, as we won't enable the on-demand feature for now on the packages api
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 3 total unresolved issues (including 2 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 24bb6c1. Configure here.
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
| // Separate connection from the API's default req.temporal client — packages_worker | ||
| // (npm/maven/pypi/osv/security-contacts/...) polls task queues in its own Temporal | ||
| // namespace (CROWD_PACKAGES_TEMPORAL_NAMESPACE), not the API's default namespace. | ||
| export function getPackagesTemporalClient(): Promise<Client> { |
| export function getPackagesTemporalClient(): Promise<Client> { | ||
| if (!_init) { | ||
| if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { | ||
| throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE') |
| export function getPackagesTemporalClient(): Promise<Client> { | ||
| if (!_init) { | ||
| if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { | ||
| throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE') | ||
| } |
| export function getPackagesTemporalClient(): Promise<Client> { | ||
| if (!_init) { | ||
| if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { | ||
| throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE') |
This pull request introduces an on-demand workflow for ingesting security contacts for a package when requested via the API, especially for packages that have not yet been processed by the regular batch job. The main changes include new workflow and activity definitions to support single-package ingestion, integration of this workflow into the public API, and refactoring to share logic between batch and single-ingest paths.
On-demand security contacts ingestion:
getPackage.tsthat, when a package's security contacts have not been refreshed, invokes an on-demand Temporal workflow to ingest contacts for that package (ondemandWorkflowId, workflow execution logic).ondemandWorkflowId).Workflow and activity changes:
ingestSecurityContactsForPurlWorkflowand supporting activityingestSecurityContactsForPurlActivityto process a single package, with appropriate timeouts and retry policies for synchronous API use [1] [2] [3] [4].Core ingestion logic refactoring:
buildBaseDeps,processRepo) to allow both batch and single-ingest paths to reuse the same code [1] [2].ingestSingle.tsto encapsulate the logic for finding the best repo for a purl and running the extraction pipeline for just that repo.Supporting changes:
These changes enable the API to provide up-to-date security contact information for any package on demand, improving responsiveness for less critical or newly added packages.
Note
Medium Risk
Public package GET can synchronously wait on Temporal and external fetches (latency/timeouts), and misconfigured packages Temporal would break on-demand ingest while the endpoint degrades to cached data.
Overview
When get package sees a repo that has never had security contacts evaluated (
contactsLastRefreshedis null), it now blocks up to ~60s on a packages-namespace Temporal workflow instead of returning empty contacts forever for non–batch-covered packages.The API gets a dedicated
getPackagesTemporalClientandpackagesTemporalconfig (CROWD_PACKAGES_TEMPORAL_*), separate from the main API Temporal client. The security-contacts worker is wired to that namespace viaCROWD_PACKAGES_TEMPORAL_NAMESPACE.Worker side: new
ingestSecurityContactsForPurlWorkflow/ activity plusingestSingle.ts(best-repo selection aligned withgetPackageDetailByPurl, nois_criticalgate).buildBaseDepsandprocessRepoare exported so batch and on-demand share the same extraction path. The scheduled batch sweep drops the GitHub-only host filter so non-GitHub repos can be retried on the normal cadence.Concurrent requests for the same purl use a deterministic workflow id with
USE_EXISTINGso only one ingest runs; failures are logged and the handler still serves cached detail.Reviewed by Cursor Bugbot for commit 24bb6c1. Bugbot is set up for automated code reviews on this repo. Configure here.